package com.fxkj.cloud.drms.infra.repository.timeseries;

import com.fxkj.cloud.dbs.common.constants.DataType;
import com.fxkj.cloud.dbs.common.model.DeviceAttributeKey;
import com.fxkj.cloud.dbs.common.model.aggregation.AggregationType;
import com.fxkj.cloud.dbs.common.model.aggregation.IntervalType;
import com.fxkj.cloud.dbs.common.model.aggregation.QueryType;
import com.fxkj.cloud.dbs.common.model.aggregation.TsQuery;
import com.fxkj.cloud.dbs.common.model.aggregation.TsQueryResult;
import com.fxkj.cloud.dbs.common.model.kv.*;
import com.fxkj.cloud.drms.domain.device.repository.*;
import com.fxkj.cloud.drms.domain.ts.repository.TimeseriesRepository;
import com.fxkj.cloud.drms.infra.config.TDengineJdbcTemplateConfig;
import com.fxkj.cloud.drms.infra.sql.TDengine.*;
import com.google.common.util.concurrent.Futures;
import com.google.common.util.concurrent.ListenableFuture;
import com.google.common.util.concurrent.MoreExecutors;
import jakarta.annotation.PostConstruct;
import lombok.Getter;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.lang3.StringUtils;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.stereotype.Component;
import org.springframework.transaction.annotation.Transactional;

import java.sql.*;
import java.time.Instant;
import java.time.LocalDateTime;
import java.time.ZoneOffset;
import java.time.temporal.ChronoUnit;
import java.util.*;
import java.util.stream.Collectors;

/**
 * <p>description: 涛思时序数据仓储实现</p>
 * @author ww
 * @date 2025/3/18 10:04
 */
@Component
@Slf4j
public class TDengineTimeseriesRepository extends TDEngineAbstractAsyncDao implements TimeseriesRepository {
    private NoSqlTsPartitionDate tsFormat;
    @Getter
    @Value("${fx.ts.tdengine.query.ts_key_value_partitioning}")
    private String partitioning;
    @Value("${fx.ts.tdengine.query.ts_key_value_partitions_max_cache_size:100000}")
    private long partitionsCacheSize;
    private TDEngineTsPartitionsCache tDengineTsPartitionsCache;

    private final Long INTERVAL = 60L;

    public TDengineTimeseriesRepository(TDEngineBufferedRateReadExecutor readExecutor, TDEngineBufferedRateWriteExecutor writeExecutor, TDengineJdbcTemplateConfig tdEngineJdbcTemplateConfig) {
        super(readExecutor, writeExecutor, tdEngineJdbcTemplateConfig);
    }

    @PostConstruct
    public void init() {
        super.startExecutor();
        Optional<NoSqlTsPartitionDate> partition = NoSqlTsPartitionDate.parse(partitioning);
        if (partition.isPresent()) {
            tsFormat = partition.get();
            if (!isFixedPartitioning() && partitionsCacheSize > 0) {
                tDengineTsPartitionsCache = new TDEngineTsPartitionsCache(partitionsCacheSize);
            }
        } else {
            log.warn("Incorrect configuration of partitioning {}", partitioning);
            throw new RuntimeException("Failed to parse partitioning property: " + partitioning + "!");
        }
    }

    @Override
    public ListenableFuture<Void> createSuperTable(UUID rootOrganizationId, UUID productId, Map<String, DataType> attributeKeyDataTypeMap) {
        PreparedStatement prepStmt = getSuperTableCreatePrepStmt(rootOrganizationId, productId, attributeKeyDataTypeMap);
        ListenableFuture<Void> future = null;
        future = getFuture(executeAsyncWrite(rootOrganizationId, prepStmt), rs -> null);
        ListenableFuture<Void> transform = Futures.transform(future, result -> null, readResultsProcessingExecutor);
        closeConnectionCallback(transform, prepStmt);
        return future;
    }

    @Override
    public ListenableFuture<Void> addProductAttribute(UUID productId, Map<String, DataType> attributeKeyDataTypeMap) {
        List<PreparedStatement> prepStmts = getAddProductAttributePrepStmt(productId, attributeKeyDataTypeMap);
        // 将并行执行改为顺序执行
        ListenableFuture<Void> chain = Futures.immediateFuture(null);
        for (PreparedStatement prepStmt : prepStmts) {
            chain = Futures.transformAsync(chain, input -> {
                ListenableFuture<Void> future = getFuture(executeAsyncWrite(productId, prepStmt), rs -> null);
                ListenableFuture<Void> transform = Futures.transform(future, result -> null, readResultsProcessingExecutor);
                closeConnectionCallback(transform, prepStmt);
                return transform;
            }, readResultsProcessingExecutor);
        }
        return chain;
    }

    @Override
    public ListenableFuture<Void> removeProductAttribute(UUID productId, Set<String> attributeKeys) {
        List<PreparedStatement> prepStmts = getRemoveProductAttributePrepStmt(productId, attributeKeys);
        // 将并行执行改为顺序执行
        ListenableFuture<Void> chain = Futures.immediateFuture(null);
        for (PreparedStatement prepStmt : prepStmts) {
            chain = Futures.transformAsync(chain, input -> {
                ListenableFuture<Void> future = getFuture(executeAsyncWrite(productId, prepStmt), rs -> null);
                ListenableFuture<Void> transform = Futures.transform(future, result -> null, readResultsProcessingExecutor);
                closeConnectionCallback(transform, prepStmt);
                return transform;
            }, readResultsProcessingExecutor);
        }
        return chain;
    }

    @Override
    @Transactional
    public void createSonTable(UUID rootOrganizationId, UUID productId, UUID deviceId) {
        PreparedStatement prepStmt = null;
        try {
            prepStmt = getSonTableCreatePrepStmt(rootOrganizationId, productId, deviceId);
        } catch (SQLException e) {
            throw new RuntimeException(e);
        }
        ListenableFuture<Void> future = null;
        future = getFuture(executeAsyncWrite(rootOrganizationId, prepStmt), rs -> null);
        ListenableFuture<Void> transform = Futures.transform(future, result -> null, readResultsProcessingExecutor);
        closeConnectionCallback(transform, prepStmt);
    }

    private ListenableFuture<Boolean> getResult(TDEngineResultSet<Boolean> rs) {
        return Futures.transform(rs.booleanResult(MoreExecutors.directExecutor(), result -> result), input -> input, MoreExecutors.directExecutor());
    }


    @Override
    public ListenableFuture<Integer> save(UUID rootOrganizationId, UUID productId, UUID deviceId, List<TsKvEntry> tsKvEntries) {
        long ts = tsKvEntries.getFirst().getTs();
        long partitionTs = toPartitionTs(ts);
        ListenableFuture<Boolean> future = null;
        //获取参数化的SQL语句
        // INSERT INTO stb(tbname,ts,id,col1,col2...) VALUES (tbname.v,ts.v,id.v,...);
        PreparedStatement prepStmt = getSavePrepStmt(productId, tsKvEntries);
        int size = tsKvEntries.size();
        //填充参数
        //填充tbname,ts,id
        try {
            prepStmt.setString(1, SqlConstants.getOrdinaryTableName(deviceId));
            prepStmt.setLong(2, ts);
            prepStmt.setString(3, SqlConstants.UUIDToString(deviceId));
            prepStmt.setLong(4, partitionTs);
        } catch (SQLException e) {
            closeConnection(prepStmt);
            throw new RuntimeException(e);
        }
        //填充其他列
        try {
            for (int offset = 0; offset < size; offset++) {
                setParams(prepStmt, offset + 5, tsKvEntries.get(offset));
            }
        } catch (SQLException e) {
            closeConnection(prepStmt);
            throw new RuntimeException(e);
        }
        future = getFuture(executeAsyncWrite(rootOrganizationId, prepStmt), rs -> null);
        ListenableFuture<Integer> transform = Futures.transform(future, result -> tsKvEntries.size(), readResultsProcessingExecutor);
        closeConnectionCallback(transform, prepStmt);
        return transform;
    }

    @Override
    public ListenableFuture<List<TsKvEntry>> getLatestTsKv(UUID deviceId, List<DeviceAttributeKey> deviceAttributeKeys) {
        PreparedStatement selectLatestPrepStmt = getSelectLatestPrepStmt(deviceId, deviceAttributeKeys);
        ListenableFuture<List<TsKvEntry>> future = null;
        Map<String, DataType> keyTypeMap = deviceAttributeKeys.stream().collect(Collectors.toMap(DeviceAttributeKey::getKey, DeviceAttributeKey::getDataType));
        future = getFutureAsync(executeAsyncRead(deviceId, selectLatestPrepStmt), rs -> convertToLatestTsKvEntries(rs, keyTypeMap));
        closeConnectionCallback(future, selectLatestPrepStmt);
        return future;
    }

    @Override
    public ListenableFuture<List<TsKvEntry>> getHistoryTsKv(UUID deviceId, List<DeviceAttributeKey> deviceAttributeKeys, long startTime, long endTime, Integer limit) {
        PreparedStatement selectLatestPrepStmt = getSelectHistoryPrepStmt(deviceId, deviceAttributeKeys, limit);
        try {
            selectLatestPrepStmt.setLong(1, startTime);
            selectLatestPrepStmt.setLong(2, endTime);
        } catch (SQLException e) {
            closeConnection(selectLatestPrepStmt);
            throw new RuntimeException(e);
        }
        ListenableFuture<List<TsKvEntry>> future = null;
        Map<String, DataType> keyTypeMap = deviceAttributeKeys.stream().collect(Collectors.toMap(DeviceAttributeKey::getKey, DeviceAttributeKey::getDataType));
        future = getFutureAsync(executeAsyncRead(deviceId, selectLatestPrepStmt), rs -> convertHistoryTsKvEntries(rs, keyTypeMap));
        closeConnectionCallback(future, selectLatestPrepStmt);
        return future;
    }

    @Override
    public ListenableFuture<List<TsKvEntry>> queryLastValueForString(UUID deviceId, List<String> attributeKeys) {
        PreparedStatement keySelectLatestPrepStmt = getKeysSelectLatestPrepStmt(deviceId, attributeKeys);
        ListenableFuture<List<TsKvEntry>> future = null;
        future = getFutureAsync(executeAsyncRead(deviceId, keySelectLatestPrepStmt), rs -> convertToLatestTsKvEntry(rs, attributeKeys));
        closeConnectionCallback(future, keySelectLatestPrepStmt);
        return future;
    }

    @Override
    public ListenableFuture<Map<UUID, Map<String, KvEntry>>> queryDevicesLastValueByProductId(Set<UUID> productIds, List<String> attributeKeys, UUID organizationId) {
        PreparedStatement prepStmt = getProductSelectLatestPrepStmt(productIds, attributeKeys);
        ListenableFuture<Map<UUID, Map<String, KvEntry>>> future = null;
        future = getFutureAsync(executeAsyncRead(organizationId, prepStmt), rs -> convertToProductLatestTsKvEntry(rs, attributeKeys));
        closeConnectionCallback(future, prepStmt);
        return future;
    }

    @Override
    public ListenableFuture<Void> saveDeviceStatus(DeviceStatus status) {
        ListenableFuture<Void> future = null;
        PreparedStatement prepStmt = getDeviceStatusPrepStmt(status.getDeviceId());
        try {
            prepStmt.setString(1, SqlConstants.UUIDToString(status.getOrganizationId()));
            prepStmt.setString(2, SqlConstants.UUIDToString(status.getProductId()));
            prepStmt.setString(3, SqlConstants.UUIDToString(status.getDeviceId()));
            prepStmt.setLong(4, status.getTs());
            prepStmt.setInt(5, status.getStatus().getCode());
        } catch (SQLException e) {
            closeConnection(prepStmt);
            throw new RuntimeException(e);
        }
        future = getFuture(executeAsyncWrite(status.getOrganizationId(), prepStmt), rs -> null);
        ListenableFuture<Void> transform = Futures.transform(future, result -> null, readResultsProcessingExecutor);
        closeConnectionCallback(transform, prepStmt);
        return transform;
    }

    @Override
    public ListenableFuture<ProductStatusStatistics> getProductStatusStatistics(UUID organizationId, Set<UUID> productIds, Long selectTime) {
        PreparedStatement prepStmt = statisticsProductLatestStatusPrepStmt(productIds.size());
        try {
            int index = 1;
            for (UUID productId : productIds) {
                prepStmt.setString(index, SqlConstants.UUIDToString(productId));
                index++;
            }
            prepStmt.setLong(index, selectTime);
        } catch (SQLException e) {
            closeConnection(prepStmt);
            throw new RuntimeException(e);
        }
        ListenableFuture<ProductStatusStatistics> future = null;
        future = getFutureAsync(executeAsyncRead(organizationId, prepStmt), this::convertToDeviceStatsStatistics);
        closeConnectionCallback(future, prepStmt);
        return future;
    }

    @Override
    public ListenableFuture<ProductDeviceStatusChangeHistory> getProductDeviceStatusChangeHistory(UUID organizationId, Set<UUID> productIds, UUID productId, UUID deviceId, Long startTime, Long endTime, Integer intervalValue, IntervalType intervalTypeEnum) {
        PreparedStatement prepStmt = getProductDeviceStatusChangeHistoryPrepStmt(productIds, productId, deviceId, intervalTypeEnum.format(intervalValue));
        try {
            if (!productIds.isEmpty()) {
                int index = 1;
                boolean allDeviceFlag = false;
                for (UUID id : productIds) {
                    //可能产品id也为空,则要走下面查询全部设备逻辑
                    if (null == id) {
                        allDeviceFlag = true;
                        break;
                    }
                    prepStmt.setString(index, SqlConstants.UUIDToString(id));
                    index++;
                }
                if (!allDeviceFlag){
                    prepStmt.setLong(index++, startTime);
                    prepStmt.setLong(index, endTime);
                }
            }
            if (productId != null) {
                prepStmt.setString(1, SqlConstants.UUIDToString(productId));
                prepStmt.setLong(2, startTime);
                prepStmt.setLong(3, endTime);
            } else {
                if (deviceId == null) {
                    prepStmt.setLong(1, startTime);
                    prepStmt.setLong(2, endTime);
                } else {
                    prepStmt.setString(1, SqlConstants.UUIDToString(deviceId));
                    prepStmt.setLong(2, startTime);
                    prepStmt.setLong(3, endTime);
                }
            }

        } catch (SQLException e) {
            closeConnection(prepStmt);
            throw new RuntimeException(e);
        }
        ListenableFuture<ProductDeviceStatusChangeHistory> future = null;
        future = getFutureAsync(executeAsyncRead(organizationId, prepStmt), this::convertToProductDeviceStatsChangeHistory);
        closeConnectionCallback(future, prepStmt);
        return future;
    }

    @Override
    public ListenableFuture<List<ProductDeviceLightStatistics>> getProductLightStatistics(UUID organizationId, Set<UUID> productIds, Long startTime, Long endTime) {
        PreparedStatement prepStmt = getProductLightStatisticsPrepStmt(productIds);
        try {
            int index = 0;
            for (UUID productId : productIds) {
                prepStmt.setLong((3 * index) + 1, startTime);
                prepStmt.setLong((3 * index) + 2, endTime);
                prepStmt.setLong((3 * index) + 3, INTERVAL);
                index++;
            }
        } catch (SQLException e) {
            closeConnection(prepStmt);
            throw new RuntimeException(e);
        }
        ListenableFuture<List<ProductDeviceLightStatistics>> future = null;
        future = getFutureAsync(executeAsyncRead(organizationId, prepStmt), this::convertToProductDeviceLightStatistics);
        closeConnectionCallback(future, prepStmt);
        return future;
    }

    @Override
    public ListenableFuture<List<DeviceStatusLatestInfo>> getDeviceLatestStatusInfo() {
        PreparedStatement prepStmt = getDeviceLatestStatusInfoPrepStmt();
        ListenableFuture<List<DeviceStatusLatestInfo>> future = null;
        future = getFutureAsync(executeAsyncRead(null, prepStmt), this::convertToDeviceStatusLatestInfo);
        closeConnectionCallback(future, prepStmt);
        return future;
    }

    @Override
    public ListenableFuture<List<TsQueryResult>> tsQuery(TsQuery query) {
        PreparedStatement prepStmt = getTsQueryPrepStmt(query);
        Long startTime = query.getStartTime();
        Long endTime = query.getEndTime();
        QueryType queryType = query.getQueryType();
        AggregationType aggregationType = query.getAggregationQuery().getAggregationType();
        List<String> queryKeys = query.getQueryKeys();
        List<UUID> queryIds = query.getQueryIds();
        try {
            for (int i = 0; i < queryIds.size(); i++) {
                prepStmt.setLong(2 * i + 1, startTime);
                prepStmt.setLong(2 * i + 2, endTime);
            }
        } catch (SQLException e) {
            closeConnection(prepStmt);
            throw new RuntimeException(e);
        }
        ListenableFuture<List<TsQueryResult>> future = null;
        future = getFutureAsync(executeAsyncRead(null, prepStmt), result -> convertToTsQueryResult(result, queryKeys, StringUtils.isNotEmpty(query.getAggregationQuery().getInterval().getIntervalString()), queryType, aggregationType));
        closeConnectionCallback(future, prepStmt);
        return future;
    }


    private boolean isFixedPartitioning() {
        return tsFormat.getTruncateUnit().equals(ChronoUnit.FOREVER);
    }

    long toPartitionTs(long ts) {
        LocalDateTime time = LocalDateTime.ofInstant(Instant.ofEpochMilli(ts), ZoneOffset.UTC);
        return tsFormat.truncatedTo(time).toInstant(ZoneOffset.UTC).toEpochMilli();
    }
}
